package org.example.cdc;

import com.ververica.cdc.connectors.postgres.PostgreSQLSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Properties;

public class PgCdc {
    public static void main(String[] args) throws Exception {
        Properties properties = new Properties();
        properties.setProperty("snapshot.mode", "never");
        properties.setProperty("debezium.slot.name", "pg_cdc");
        properties.setProperty("debezium.slot.drop.on.stop", "true");
        properties.setProperty("include.schema.changes", "true");

        SourceFunction<String> sourceFunction = PostgreSQLSource.<String>builder()
                .hostname("localhost")
                .port(5432)
                .database("postgres") // monitor postgres database
                .schemaList("public")  // monitor inventory schema
                .tableList("public.test") // monitor products table
                .username("postgres")
                .password("hyh.5795")
                .decodingPluginName("pgoutput") // 使用解码插件, pgoutput 是 PostgreSQL 10+ 推荐的插件
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .debeziumProperties(properties)
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> stream = env.addSource(sourceFunction)
                .assignTimestampsAndWatermarks(WatermarkStrategy.noWatermarks());


       stream.print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

        env.execute();
    }
}
